/*
 * Copyright (c) 2020-2099 the original author or authors. All rights reserve.
 */
package org.tianyun.cloud.amqp.rabbitmq.factory;

import org.aopalliance.aop.Advice;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.util.StringUtils;
import org.tianyun.cloud.amqp.rabbitmq.advice.ListenerInvokeAdvice;
import org.tianyun.cloud.amqp.rabbitmq.configuration.RabbitMqProperties;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

/**
 * 受托管的SimpleRabbitListenerContainerFactory
 *
 * @auther ebert_chan
 */
public class ManagedRabbitListenerContainerFactory implements RabbitListenerContainerFactory<SimpleMessageListenerContainer>, ApplicationContextAware,
        ApplicationEventPublisherAware {

    private RabbitMqProperties properties;

    private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;

    private ApplicationContext applicationContext;

    private SimpleRabbitListenerContainerFactory delegate;

    private final Map<String, RabbitListenerEndpoint> rabbitListenerEndpoints;

    private final Map<RabbitListenerEndpoint, SimpleMessageListenerContainer> endpointListenerContainers;

    private final List<SimpleMessageListenerContainer> nonEndPointListenerContainers;

    private final Object monitor = new Object();

    public ManagedRabbitListenerContainerFactory(RabbitMqProperties properties, RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry, ObjectProvider<List<ListenerInvokeAdvice>> adviceChain) {
        this.rabbitListenerEndpointRegistry = rabbitListenerEndpointRegistry;
        this.properties = properties;

        delegate = new SimpleRabbitListenerContainerFactory();

        rabbitListenerEndpoints = new HashMap<>();
        endpointListenerContainers = new HashMap<>();
        nonEndPointListenerContainers = new ArrayList<>();
        if (adviceChain != null) {
            this.setAdviceChain(adviceChain.getIfAvailable());
        }
        this.configure(properties);
    }

    private void configure(RabbitMqProperties properties) {
        delegate.setPrefetchCount(properties.getPrefetch());
        delegate.setConcurrentConsumers(properties.getConcurrency());
        delegate.setAcknowledgeMode(properties.getAcknowledgeMode());
        delegate.setMissingQueuesFatal(false);
    }

    public void setAdviceChain(List<ListenerInvokeAdvice> adviceChain) {
        if (adviceChain != null && !adviceChain.isEmpty()) {
            delegate.setAdviceChain(adviceChain.stream().toArray(Advice[]::new));
        }
    }

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        delegate.setConnectionFactory(connectionFactory);
    }

    public void setMessageConverter(MessageConverter messageConverter) {
        delegate.setMessageConverter(messageConverter);
    }

    @Override
    public SimpleMessageListenerContainer createListenerContainer() {
        SimpleMessageListenerContainer container;
        synchronized (this.monitor) {
            container = delegate.createListenerContainer();
            nonEndPointListenerContainers.add(container);
        }
        return container;
    }

    @Override
    public SimpleMessageListenerContainer createListenerContainer(RabbitListenerEndpoint endpoint) {
        SimpleMessageListenerContainer container;
        synchronized (this.monitor) {
            container = delegate.createListenerContainer(endpoint);
            if (endpoint != null) {
                if (!rabbitListenerEndpoints.containsKey(endpoint.getId())) {
                    rabbitListenerEndpoints.put(endpoint.getId(), endpoint);
                }
                endpointListenerContainers.put(endpoint, container);
            } else {
                nonEndPointListenerContainers.add(container);
            }
        }
        return container;
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.delegate.setApplicationContext(applicationContext);
    }

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        delegate.setApplicationEventPublisher(applicationEventPublisher);
    }

    public void refresh() {
        synchronized (this.monitor) {
            destroyNonEndpointListenerContainer();
            destroyEndpointListenerContainer();
            registerListenerContainer();
        }
    }

    public void refresh(RabbitMqProperties properties) {
        this.configure(properties);
        this.refresh();
    }

    private void registerListenerContainer() {
        if (rabbitListenerEndpoints.isEmpty()) {
            return;
        }
        Iterator<Entry<String, RabbitListenerEndpoint>> iterator = rabbitListenerEndpoints.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, RabbitListenerEndpoint> entry = iterator.next();
            RabbitListenerEndpoint rabbitListenerEndpoint = entry.getValue();
            rabbitListenerEndpointRegistry.registerListenerContainer(rabbitListenerEndpoint, this, true);
        }
    }

    private void destroyEndpointListenerContainer() {
        if (endpointListenerContainers.isEmpty()) {
            return;
        }

        Iterator<Entry<RabbitListenerEndpoint, SimpleMessageListenerContainer>> iterator = endpointListenerContainers.entrySet().iterator();

        while (iterator.hasNext()) {
            Entry<RabbitListenerEndpoint, SimpleMessageListenerContainer> entry = iterator.next();
            RabbitListenerEndpoint endpoint = entry.getKey();
            SimpleMessageListenerContainer container = entry.getValue();

            if (StringUtils.hasText(endpoint.getGroup()) && this.applicationContext.containsBean(endpoint.getGroup())) {
                @SuppressWarnings("unchecked")
                List<MessageListenerContainer> containerGroup = this.applicationContext.getBean(endpoint.getGroup(), List.class);
                containerGroup.remove(container);
            }

            rabbitListenerEndpointRegistry.unregisterListenerContainer(endpoint.getId());

            container.destroy();

            iterator.remove();
        }
    }

    private void destroyNonEndpointListenerContainer() {
        if (nonEndPointListenerContainers.isEmpty()) {
            return;
        }
        Iterator<SimpleMessageListenerContainer> iterator = nonEndPointListenerContainers.iterator();
        while (iterator.hasNext()) {
            SimpleMessageListenerContainer container = iterator.next();
            container.destroy();
            iterator.remove();
        }
    }

}
